package com.example.mqproduct.consumer;

import com.alibaba.fastjson.JSON;
import com.example.mqproduct.config.MqConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.text.SimpleDateFormat;
import java.util.Date;

@Slf4j
@Component
public class Consumer {

    @PostConstruct
    void init() {
        messageListener();
        delayTestListener();
    }

    private void messageListener() {
        log.info("mq消费者启动...");
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(MqConfig.group);
        consumer.setNamesrvAddr(MqConfig.nameServ);
        try {
            // 订阅PushTopic下Tag为push的消息,都订阅消息
            consumer.subscribe(MqConfig.topic, "*");

            //消费模式:一个新的订阅组第一次启动从队列的最后位置开始消费 后续再启动接着上次消费的进度开始消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            //负载均衡模式消费
            // consumer.setMessageModel(MessageModel.BROADCASTING);
            //可以修改每次消费消息的数量，默认设置是每次消费一条
            consumer.setConsumeMessageBatchMaxSize(1);
            //在此监听中消费信息，并返回消费的状态信息
            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
                // 会把不同的消息分别放置到不同的队列中
                for (MessageExt msg : msgs) {
                    log.info("接受tag：{}", msg.getTags());
                    log.info("接收到了消息：{}", new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            consumer.start();
            log.info("mq消费者启动完成...");
        } catch (Exception e) {
            log.info("mq消费者启动异常：{}", e.getLocalizedMessage());
        }
    }

    /**
     * 延迟 消费
     */
    public void delayTestListener() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("delayGroup");
        consumer.setNamesrvAddr(MqConfig.nameServ);
        try {
            // 订阅PushTopic下Tag为push的消息,都订阅消息
            consumer.subscribe("delayPushMsg", "*");

            // 程序第一次启动从消息队列头获取数据
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            //可以修改每次消费消息的数量，默认设置是每次消费一条
            consumer.setConsumeMessageBatchMaxSize(1);
            //在此监听中消费信息，并返回消费的状态信息
            SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.sss");
            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
                // 会把不同的消息分别放置到不同的队列中
                for (MessageExt msg : msgs) {
                    log.info("接收时间: {}", format.format(new Date()));
                    log.info("Receive message:msgId={},msgBody={},delay={} ms",
                            msg.getMsgId(),
                            new String(msg.getBody()),
                            (System.currentTimeMillis() - msg.getStoreTimestamp()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            consumer.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
